package com.chb.pagerank;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MyRunJob {
	public static enum MyCounter {
		countName;
	}
	public static void main(String[] args) throws Exception {
		System.setProperty("HADOOP_USER_NAME", "chb");
		Configuration conf = new Configuration();
		int i =0;//记录统计的次数
		//递归收敛的差值
		double d = 0.01;
		while (true) {//递归计算
			conf.setInt("runCount", i);
			FileSystem fs = FileSystem.get(conf);
			Job job = Job.getInstance();
			job.setJar("C:\\Users\\12285\\Desktop\\pr.jar");
			job.setJobName("pr"+i);
			job.setJarByClass(MyRunJob.class);

			job.setMapperClass(MyPageRankMapper.class);
			job.setReducerClass(MyPageRankReducer.class);

			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(Text.class);
			//设置拆分
			job.setInputFormatClass(KeyValueTextInputFormat.class);
			//第一轮输入
			Path in = new Path("/user/chb/input/pagerank.txt");
			if (i > 0) {
				in = new Path("/user/chb/output/pr"+(i-1));
			}
			FileInputFormat.addInputPath(job, in);

			Path  out = new Path("/user/chb/output/pr"+i);
			if (fs.exists(out)) {
				fs.delete(out, true);
			}
			FileOutputFormat.setOutputPath(job, out);
			boolean f = job.waitForCompletion(true);
			if (f) {//一轮job执行完成
				//每次的执行结果，对比新旧的PR是否达到了收敛
				long sum = job.getCounters().findCounter(MyCounter.countName).getValue();
				System.out.println(sum);
				/**
				 * 因为每次的reducer中按照key进行四次计算， 每次计算增加一次j
				 * reducer:
				 * 		context.getCounter(MyRunJob.MyCounter.countName).increment(j);
				 * 所以求取平均差值。要除以4*1000.0
				 * 		在reducer中：int j = (int) (d*1000.0);
				 */
				double avgD = sum/4000.0;
				if (avgD < d) {//达到了收敛
					break;
				}
				System.out.println("第"+i+"次job"+"执行完成");
				i++;
			}
		}

	}
}
